package com.atguigu.tingshu.account.listener;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.account.service.UserAccountService;
import com.atguigu.tingshu.common.kafka.constant.KafkaConst;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 用户注册消息监听器
 * 监听用户注册消息，进行账户初始化
 */
@Slf4j
@Component
public class UserRegisterListener {

    @Autowired
    private UserAccountService userAccountService;

    /**
     * 监听用户注册消息
     *
     * @param record 消息记录
     * @param ack    手动确认对象
     */
    @KafkaListener(topics = KafkaConst.TOPIC_USER_REGISTER, groupId = KafkaConst.GROUP_USER)
    public void handleUserRegister(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            // 1. 获取消息内容
            String message = record.value();
            log.info("收到用户注册消息 - Topic: {}, Partition: {}, Offset: {}, Message: {}",
                    record.topic(), record.partition(), record.offset(), message);
            // 2. 解析消息
            Map<String, Object> registerMessage = JSON.parseObject(message, Map.class);
            Long userId = Long.valueOf(registerMessage.get("userId").toString());
            String nickname = (String) registerMessage.get("nickname");
            log.info("解析用户注册消息 - userId: {}, nickname: {}", userId, nickname);
            // 3. 初始化用户账户
            userAccountService.initUserAccount(userId);
            // 4. 手动确认消息
            ack.acknowledge();
            log.info("用户注册消息处理成功 - userId: {}, Offset: {}", userId, record.offset());
        } catch (Exception e) {
            log.error("处理用户注册消息失败 - Offset: {}, Error: {}",
                    record.offset(), e.getMessage(), e);
            // 注意：不调用 ack.acknowledge()，消息会重新消费
            // 如果需要跳过错误消息，可以调用 ack.acknowledge()
        }
    }

}
